package sentimentalAnalyzer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.TreeMap;
import java.util.Map.Entry;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class SentimentalMapper extends Mapper<LongWritable, Text, Text, Text> {
	
	static TreeMap<String, WordStruct> dict = new TreeMap<String, WordStruct>();
	
	public void map(LongWritable key, Text value, Context context)
		throws IOException, InterruptedException {
		
		String sentence = value.toString().toLowerCase();
		Pattern p = Pattern.compile("[a-z]+");
		Matcher m = p.matcher(sentence);
		 
		ArrayList<String> wdList = new ArrayList<String>();
		
		while(m.find()){
			String s = m.group();
			wdList.add(s);
		}		

		ArrayList<Double> al = calculate(wdList, context);
		if (al == null) return;
		double posV = al.get(0);
		double negV = al.get(1);
		
		context.write(new Text(String.valueOf(posV) + "#" + String.valueOf(negV)), new Text(value.toString()));
	}
	
	ArrayList<Double> calculate(ArrayList<String> list, Context c) throws IOException, InterruptedException{
		
		double posSum = 0.0;
		double negSum = 0.0;
		int cnt = 0;
		
		double posLevel = -1.0;
		double negLevel = -1.0;
		
		boolean negFlag = false;
		
		for (int i = 0; i < list.size(); i ++){
			for (int l = 1; l <= 3; l ++){
				if (i + l - 1 >= list.size()) continue;
				String term = new String("");
				for (int j = i; j <= i+l-1; j ++){
					term += list.get(j) + "_";
				}
				term = term.substring(0, term.length()-1);
				
				if (term.equals("no") || term.equals("not") || term.equals("non") || term.equals("don't")){
					negFlag = true;
					continue;
				}
				
				//current term's value?
				if (!dict.containsKey(term))
					continue;
				
				Entry<String, WordStruct> e = dict.ceilingEntry(term);
				if (e.getValue().POS.equals("r")){
					posLevel = e.getValue().posScore + 1.0;
					negLevel = e.getValue().negScore + 1.0;
					continue;
				}
				
				double tPosScore = e.getValue().posScore;
				double tNegScore = e.getValue().negScore;
				if (negFlag){
					tNegScore *= e.getValue().posScore;
					tPosScore *= e.getValue().negScore;
					if (posLevel != -1.0 || negLevel != -1.0){
						double temp = posLevel;
						posLevel = negLevel;
						negLevel = temp;
					}
					negFlag = false;
				}
				if (posLevel != -1.0){
					posSum += tPosScore * posLevel;
					posLevel = -1.0;
				}
				else
					posSum += tPosScore;
				
				if (negLevel != -1.0){
					negSum += tNegScore * negLevel;
					negLevel = -1.0;
				}
				else
					negSum += tNegScore;
				
				cnt ++;
			}
		}
		if (cnt == 0)
			return null;
		ArrayList<Double> al = new ArrayList<Double>();
		al.add(posSum / cnt);
		al.add(negSum / cnt);
		
		return al;
	}
	
	protected void setup(Context context) throws IOException, InterruptedException{
		String emotionIndexInput = "/user/shuqingya/sentimental/emotionIndex/part-r-00000";
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(conf);
		FSDataInputStream is = fs.open(new Path(emotionIndexInput));
		
		byte[] buffer = new byte[1024*1024];
		int len = is.read(buffer);
		String index = new String(buffer, 0, len);
		
		Pattern p = Pattern.compile(".*?\n");
		Matcher m = p.matcher(index);
		while(m.find()){
			String wstr = m.group();
			WordStruct w = new WordStruct(wstr);
			wstr = wstr.split("\t")[0];
			dict.put(wstr, w);
		}
	}
	
	class WordStruct{
		String word;
		String POS;
		double posScore;
		double negScore;
		
		WordStruct(String line){
			String str[] = line.split("\t");
			word = str[0];
			String inf[] = str[1].split("#");
			POS = inf[0];
			posScore = Double.valueOf(inf[1]);
			negScore = Double.valueOf(inf[2]);
		}
	}
}
